#pragma once
#include "MessagePubSubTypes.h"
#include "ComDefine.h"
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/subscriber/DataReaderListener.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>

#include <vector>
#include <fstream>
#include <condition_variable>
#include <mutex>
#include <atomic>
#include <thread>
#include <unordered_map>
class LogMsg {
public:
    ~LogMsg();
    void init(std::string logFile);
    void appendMsg(const Message& msg);
    void run();
    void stop();
private:
    std::ofstream ofs;
    std::string strMsg;
    std::mutex mutex_;
    std::condition_variable cond_;
    std::atomic_bool running_;
};

class MessageSubscriber
{
public:

    MessageSubscriber();

    virtual ~MessageSubscriber();
    virtual bool init(InitParam param) = 0;
    //!RUN the subscriber
    void run(std::string logFile);

    //!Run the subscriber until number samples have been received.
    void run(
        uint32_t number,
        std::string logFile);
    void stop();
protected:

    eprosima::fastdds::dds::DomainParticipant* participant_;

    eprosima::fastdds::dds::TypeSupport type_;

    class SubListener : public eprosima::fastdds::dds::DataReaderListener
    {
    public:

        SubListener()
            : matched_(0)
            , samples_(0), zeroCopy_(false)
        {
        }

        ~SubListener() override
        {
        }
        void runLog() {
            std::thread t(&LogMsg::run, &logMsg_);
            thread_.swap(t);
        }
        void stopLog() {
            logMsg_.stop();
            thread_.join();
        }
        void on_data_available(
            eprosima::fastdds::dds::DataReader* reader) override;

        void on_subscription_matched(
            eprosima::fastdds::dds::DataReader* reader,
            const eprosima::fastdds::dds::SubscriptionMatchedStatus& info) override;

        Message message_;

        int matched_;

        uint32_t samples_;
        bool zeroCopy_;
        LogMsg logMsg_;
    private:
        void zeroCopy(eprosima::fastdds::dds::DataReader* reader);
        void genericCopy(eprosima::fastdds::dds::DataReader* reader);
    private:
        std::thread thread_;
    };
    eprosima::fastdds::dds::Subscriber* subscriber_ = nullptr;
    eprosima::fastdds::dds::Topic* topic_ = nullptr;
    SubListener* listener_ = nullptr;
    eprosima::fastdds::dds::DataReader* reader_ = nullptr;
private:
    std::thread thread_;
};

class MessageSubscriber_tcp : public MessageSubscriber
{
public:
    virtual bool init(InitParam param);
};
class MessageSubscriber_udp : public MessageSubscriber
{
public:
    virtual bool init(InitParam param);
};
class MessageSubscriber_shm : public MessageSubscriber
{
public:
    virtual bool init(InitParam param);
};
class MessageSubscriber_ds : public MessageSubscriber
{
public:
    virtual bool init(InitParam param);
};
